#include "utils.h"
#include "ThreadQueue.hpp"
#include "ipc_hv_soa.h"
#include "local_registry_common.hpp"

struct ipc_hv_soa_remote_client_data
{
    uint32_t service_id;
    uint32_t msg_type;
    uint32_t msg_seqid;
    uint32_t client_id;
    bool is_complete;
    uint32_t data_len;
    void *data;
};

struct ipc_hv_soa_remote_client
{
    uint32_t client_id;                // 客户端ID(服务注册中心分配)
    std::string client_name;           // 客户端进程名
    std::string client_localaddr1;     // remote client listen socket
    local_client_status client_status; // 客户端进程运行状态
    hio_t *client_send_io;             // send
    hio_t *client_recv_io;             // recv

    std::atomic_uint32_t send_msg_seqid;
    std::list<ipc_hv_soa_remote_client_data> send_msg_list;
    int32_t send_msg_cond_ret;
    std::mutex send_msg_mutex;
    std::condition_variable send_msg_cond;
};

struct ipc_hv_soa_service
{
    uint32_t service_id;
    uint32_t service_type;
    uint32_t service_status;
    void *service_handler;
    std::shared_ptr<ipc_hv_soa_remote_client> provider;
    std::unordered_map<uint32_t, std::shared_ptr<ipc_hv_soa_remote_client>> listeners;
};

struct ipc_hv_soa_client
{
    hloop_t *daemon_loop;
    std::thread daemon_loop_thread;
    hio_t *daemon_io; // LOCAL_REGISTEY_SOCKET_FMT
    hio_t *listen_io; // LOCAL_REGISTEY_SOCKET_FMT1

    uint32_t daemon_cond_msgid;
    void *daemon_cond_msgdata;
    int32_t daemon_cond_ret;
    std::mutex daemo_mutex;
    std::condition_variable daemon_cond;

    hloop_t **worker_loops;
    std::vector<std::thread> worker_threads;
    std::mutex m_worker_mutex;

    std::thread msg_handler_thread;
    UnBoundedQueue<ipc_hv_soa_remote_client_data> msg_handler_queue;

    uint32_t client_id;                // 客户端ID(服务注册中心分配)
    pid_t client_pid;                  // 客户端进程ID
    std::string client_name;           // 客户端进程名
    std::string client_localaddr;      // daemon connect socket
    std::string client_localaddr1;     // listen socket
    local_client_status client_status; // 客户端进程运行状态
};

static bool g_init_flag = false;
static std::shared_ptr<ipc_hv_soa_client> g_client = nullptr;
static unpack_setting_t g_daemon_unpack_setting = {};
static unpack_setting_t g_remote_unpack_setting = {};

static std::mutex g_remote_clients_map_mutex;
static std::unordered_map<uint32_t, std::shared_ptr<ipc_hv_soa_remote_client>> g_remote_clients_map;

static std::mutex g_services_map_mutex;
static std::unordered_map<uint32_t, std::shared_ptr<ipc_hv_soa_service>> g_services_map;

static hloop_t *get_idle_loop();
static void on_close(hio_t *io);
static void on_recv_daemon(hio_t *io, void *buf, int readbytes);
static void on_recv_remote(hio_t *io, void *buf, int readbytes);
static void on_write(hio_t *io, const void *buf, int writebytes);
static void on_connect(hio_t *io);
static void on_post_event_cb(hevent_t *ev);
static void on_accept(hio_t *io);

// g_remote_clients_map operations
static std::shared_ptr<ipc_hv_soa_remote_client> find_remote_client(uint32_t client_id);                          // get from local cache
static std::shared_ptr<ipc_hv_soa_remote_client> save_remote_client(uint32_t client_id, std::string client_name); // save new remote client to local cache
static std::shared_ptr<ipc_hv_soa_remote_client> get_remote_client(uint32_t client_id);                           // get from local cache, if not find, get clientinfo from daemon

// g_services_map operations
static std::shared_ptr<ipc_hv_soa_service> find_service(uint32_t service_id);                                                                                                                          // get from local cache
static std::shared_ptr<ipc_hv_soa_service> save_service(uint32_t service_id, uint32_t service_type, uint32_t service_status, void *service_handler, std::shared_ptr<ipc_hv_soa_remote_client> client); // save new service to local cache
static std::shared_ptr<ipc_hv_soa_service> get_service(uint32_t service_id);                                                                                                                           // get from local cache, if not find, get service from daemon
static std::shared_ptr<ipc_hv_soa_service> get_service_listener(uint32_t service_id);

static int32_t connect_with_daemon();
static int32_t connect_with_remote_client(std::shared_ptr<ipc_hv_soa_remote_client> dest);
static int32_t send_msg_to_daemon(hio_t *daemon_io, uint32_t msg_id, const void *msgdata, const pb_msgdesc_t *fileds, uint32_t field_size);
static int32_t send_msg_to_client(std::shared_ptr<ipc_hv_soa_remote_client> dest, uint32_t client_id, uint32_t msg_seqid, uint32_t msg_type, uint32_t service_id, uint32_t msgdata_len, const void *msgdata);
static int32_t send_msg_to_client_sync(std::shared_ptr<ipc_hv_soa_remote_client> dest, uint32_t client_id, uint32_t msg_seqid, uint32_t service_id, uint32_t msgdata_len, const void *msgdata, void *method_resp_data, uint32_t *method_resp_data_len, uint32_t timeout_ms);

static int32_t ipc_hv_soa_inn_sync_complete(uint32_t service_id, uint32_t msg_seqid, void **method_resp_data, uint32_t method_resp_data_len);
static int32_t ipc_hv_soa_inn_trigger_to_client(std::shared_ptr<ipc_hv_soa_service> service, void *event_data, uint32_t event_data_len, std::shared_ptr<ipc_hv_soa_remote_client> client);
static int32_t register_client_req();

static hloop_t *get_idle_loop()
{
    std::lock_guard<std::mutex> worker_lock(g_client->m_worker_mutex);
    static size_t s_cur_index = 0;
    if (++s_cur_index >= LOCAL_REGISTRY_HV_LOOP_NUM_MAX)
    {
        s_cur_index = 0;
    }
    return g_client->worker_loops[s_cur_index % LOCAL_REGISTRY_HV_LOOP_NUM_MAX];
}

static void on_close(hio_t *io)
{
    LOG_PRINT_DEBUG("on_close fd[%d] error[%d]", hio_fd(io), hio_error(io));

    if (nullptr != g_client->daemon_io && hio_fd(io) == hio_fd(g_client->daemon_io))
    {
        {
            std::unique_lock<std::mutex> daemo_lock(g_client->daemo_mutex);
            g_client->daemon_cond_ret = INT32_MAX;
            g_client->daemon_io = nullptr;
            g_client->client_status = local_client_status_offline;
            g_client->daemon_cond.notify_all();
        }

        {
            std::lock_guard<std::mutex> remote_clients_map_lock(g_remote_clients_map_mutex);
            g_remote_clients_map.erase(g_client->client_id);
        }

        // todo reconnect;
        LOG_PRINT_ERROR("disconnect with daemon!");
    }
    else if (nullptr != g_client->listen_io && hio_fd(io) == hio_fd(g_client->listen_io))
    {
        LOG_PRINT_ERROR("listen socket closed!");
        g_client->listen_io = nullptr;
    }
    else
    {
        uint32_t client_id = *static_cast<uint32_t *>(hevent_userdata(io));
        std::shared_ptr<ipc_hv_soa_remote_client> client = find_remote_client(client_id);
        if (nullptr == client)
        {
            LOG_PRINT_ERROR("invalid client[%d]!", client_id);
        }
        else
        {
            if ((nullptr != client->client_send_io && hio_fd(io) == hio_fd(client->client_send_io)))
            {
                LOG_PRINT_ERROR("disconnect with remote client[%s] not send!", client->client_name.c_str());
                std::unique_lock<std::mutex> send_msg_lock(client->send_msg_mutex);
                client->send_msg_seqid = 0;
                client->send_msg_list.clear();
                client->send_msg_cond_ret = INT32_MAX;
                client->client_send_io = nullptr;
                client->client_status = local_client_status_offline;
                client->send_msg_cond.notify_all();
            }
            else if ((nullptr != client->client_recv_io && hio_fd(io) == hio_fd(client->client_recv_io)))
            {
                client->client_status = local_client_status_offline;
                client->client_recv_io = nullptr;
                LOG_PRINT_ERROR("disconnect with remote client[%s] not recv!", client->client_name.c_str());
            }
            else
            {
                client->client_status = local_client_status_offline;
                LOG_PRINT_ERROR("disconnect with client[%s]!", client->client_name.c_str());
            }
        }
    }
}

static void on_recv_daemon(hio_t *io, void *buf, int readbytes)
{
    LOG_PRINT_INFO("on_recv_daemon fd[%d], readbytes[%d]", hio_fd(io), readbytes);
    if (readbytes < (int)LOCAL_REGISTRY_HV_MSG_HEADER_SIZE)
    {
        LOG_PRINT_ERROR("not a complete msg");
        return;
    }

    bool status = false;
    uint32_t real_readbytes = (uint32_t)((uint32_t)readbytes - LOCAL_REGISTRY_HV_MSG_HEADER_SIZE);
    st_local_msg_header *recv_msg_header = (st_local_msg_header *)buf;
    if (real_readbytes != recv_msg_header->msg_len)
    {
        LOG_PRINT_ERROR("msg_len[%u] != real_readbytes[%u]", recv_msg_header->msg_len, real_readbytes);
        return;
    }

    LOG_PRINT_INFO("on_recv_daemon msg_id[%u], msg_len[%u]", recv_msg_header->msg_id, recv_msg_header->msg_len);

    void *resp = nullptr;
    int32_t ret = 0;
    uint8_t *buffer = (uint8_t *)buf + LOCAL_REGISTRY_HV_MSG_HEADER_SIZE;
    pb_istream_t stream = pb_istream_from_buffer(buffer, recv_msg_header->msg_len);
    switch (recv_msg_header->msg_id)
    {
        case local_registry_msg_register_client:
            {
                resp = calloc(1, sizeof(st_register_client_resp));
                if (nullptr == resp)
                {
                    LOG_PRINT_ERROR("calloc fail!, errno[%d]", errno);
                    ret = -1;
                }
                else
                {
                    *(st_register_client_resp *)resp = st_register_client_resp_init_zero;
                    status = pb_decode(&stream, st_register_client_resp_fields, resp);
                    if (!status)
                    {
                        LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                        free(resp);
                        resp = nullptr;
                        ret = -1;
                    }
                    else
                    {
                        st_register_client_resp *p_resp = (st_register_client_resp *)resp;
                        LOG_PRINT_DEBUG("register client id client_id[%u], client_pid[%u]!", p_resp->client_id, p_resp->client_pid);
                        ret = 0;
                    }
                }
            }
            break;
        case local_registry_msg_get_client:
            {
                resp = calloc(1, sizeof(st_get_client_resp));
                if (nullptr == resp)
                {
                    LOG_PRINT_ERROR("calloc fail!, errno[%d]", errno);
                    ret = -1;
                }
                else
                {
                    *(st_get_client_resp *)resp = st_get_client_resp_init_zero;
                    status = pb_decode(&stream, st_get_client_resp_fields, resp);
                    if (!status)
                    {
                        LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                        free(resp);
                        resp = nullptr;
                        ret = -1;
                    }
                    else
                    {
                        st_get_client_resp *p_resp = (st_get_client_resp *)resp;
                        if (p_resp->has_responser_client)
                        {
                            LOG_PRINT_DEBUG("get_client resp client_id[%u], client_name[%s]!", p_resp->responser_client.client_id, p_resp->responser_client.client_name);
                        }
                        ret = 0;
                    }
                }
            }
            break;
        case local_registry_msg_get_listener:
            {
                resp = calloc(1, sizeof(st_get_listener_resp));
                if (nullptr == resp)
                {
                    LOG_PRINT_ERROR("calloc fail!, errno[%d]", errno);
                    ret = -1;
                }
                else
                {
                    *(st_get_listener_resp *)resp = st_get_listener_resp_init_zero;
                    status = pb_decode(&stream, st_get_listener_resp_fields, resp);
                    if (!status)
                    {
                        LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                        free(resp);
                        resp = nullptr;
                        ret = -1;
                    }
                    else
                    {
                        ret = 0;
                    }
                }
            }
            break;
        case local_registry_msg_get_service:
            {
                resp = calloc(1, sizeof(st_get_service_resp));
                if (nullptr == resp)
                {
                    LOG_PRINT_ERROR("calloc fail!, errno[%d]", errno);
                    ret = -1;
                }
                else
                {
                    *(st_get_service_resp *)resp = st_get_service_resp_init_zero;
                    status = pb_decode(&stream, st_get_service_resp_fields, resp);
                    if (!status)
                    {
                        LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                        free(resp);
                        resp = nullptr;
                        ret = -1;
                    }
                    else
                    {
                        ret = 0;
                    }
                }
            }
            break;

        case local_registry_msg_service_change_status:
            {
                resp = calloc(1, sizeof(st_service_change_status));
                if (nullptr == resp)
                {
                    LOG_PRINT_ERROR("calloc fail!, errno[%d]", errno);
                }
                else
                {
                    *(st_service_change_status *)resp = st_service_change_status_init_zero;
                    status = pb_decode(&stream, st_service_change_status_fields, resp);
                    if (!status)
                    {
                        LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                    }
                    else
                    {
                        st_service_change_status *change_status = (st_service_change_status *)resp;
                        if (change_status->has_service)
                        {
                            LOG_PRINT_DEBUG("service change status service_id[%u]-service_status[%d]",
                                            change_status->service.service_id, change_status->service.service_status);
                            auto it = find_service(change_status->service.service_id);
                            if (nullptr == it)
                            {
                                LOG_PRINT_ERROR("service_id[%u] not found", change_status->service.service_id);
                            }
                            else
                            {
                                std::shared_ptr<ipc_hv_soa_remote_client> client = nullptr;
                                it->service_status = change_status->service.service_status;
                                if (change_status->has_provider_client)
                                {
                                    LOG_PRINT_DEBUG("service change status provider client_id[%u]-client_name[%s]-client_status[%d]",
                                                    change_status->provider_client.client_id,
                                                    change_status->provider_client.client_name,
                                                    change_status->provider_client.client_status);
                                    auto it1 = find_remote_client(change_status->provider_client.client_id);
                                    if (nullptr == it1)
                                    {
                                        client = save_remote_client(change_status->provider_client.client_id, change_status->provider_client.client_name);
                                    }
                                    else
                                    {
                                        client = it1;
                                    }
                                    it->provider = client;
                                }

                                if (nullptr != client && E_IPC_HV_SOA_SERVICE_TYPE_METHOD == it->service_type && nullptr != it->service_handler)
                                {
                                    PF_IPC_HV_SOA_SERVICE_STATUS_CB cb = (PF_IPC_HV_SOA_SERVICE_STATUS_CB)it->service_handler;
                                    cb(it->service_id, change_status->service.service_status);
                                }
                            }
                        }
                        else
                        {
                            LOG_PRINT_ERROR("service change status no service");
                        }
                    }
                    free(resp);
                }
            }
            return;
            // break;
        case local_registry_msg_listener_change_to_provider:
            {
                resp = calloc(1, sizeof(st_listener_change_to_provider));
                if (nullptr == resp)
                {
                    LOG_PRINT_ERROR("calloc fail!, errno[%d]", errno);
                }
                else
                {
                    *(st_listener_change_to_provider *)resp = st_listener_change_to_provider_init_zero;
                    status = pb_decode(&stream, st_listener_change_to_provider_fields, resp);
                    if (!status)
                    {
                        LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                    }
                    else
                    {
                        std::shared_ptr<ipc_hv_soa_remote_client> client = nullptr;
                        st_listener_change_to_provider *event_listen = (st_listener_change_to_provider *)resp;

                        LOG_PRINT_DEBUG("service_id[%u]-listener_clients_count[%u]-reg[%s]",
                                        event_listen->service_id, event_listen->listener_clients_count, event_listen->reg ? "true" : "false");
                        for (size_t i = 0; i < event_listen->listener_clients_count; i++)
                        {
                            LOG_PRINT_DEBUG("listener_client client_id[%u]", event_listen->listener_clients[i].client_id);
                        }

                        auto it = find_service(event_listen->service_id);
                        if (nullptr == it)
                        {
                            LOG_PRINT_ERROR("service_id[%u] not found", event_listen->service_id);
                        }
                        else
                        {
                            uint32_t i = 0;
                            for (i = 0; i < event_listen->listener_clients_count; i++)
                            {
                                st_local_client_item *p_client = &(event_listen->listener_clients[i]);
                                auto it1 = find_remote_client(p_client->client_id);
                                if (nullptr == it1)
                                {
                                    client = save_remote_client(p_client->client_id, p_client->client_name);
                                }
                                else
                                {
                                    client = it1;
                                }

                                if (event_listen->reg)
                                {
                                    auto it2 = it->listeners.find(client->client_id);
                                    if (it->listeners.end() == it2)
                                    {
                                        it->listeners.insert({client->client_id, client});
                                    }

                                    if (nullptr != it->service_handler)
                                    {
                                        void *event_data = nullptr;
                                        uint32_t event_data_len = 0;
                                        PF_IPC_HV_SOA_EVENT_LISTEN_CB listener_cb = (PF_IPC_HV_SOA_EVENT_LISTEN_CB)it->service_handler;
                                        if (IPC_HV_SOA_RET_SUCCESS == listener_cb(it->service_id, &event_data, &event_data_len))
                                        {
                                            ret = ipc_hv_soa_inn_trigger_to_client(it, event_data, event_data_len, client);
                                            if (IPC_HV_SOA_RET_SUCCESS != ret)
                                            {
                                                LOG_PRINT_ERROR("ipc_hv_soa_inn_trigger_to_client fail, ret[%d]!", ret);
                                            }

                                            if (nullptr != event_data)
                                            {
                                                free(event_data);
                                            }
                                        }
                                    }
                                }
                                else
                                {
                                    it->listeners.erase(client->client_id);
                                    break;
                                }
                            }
                        }
                    }
                    free(resp);
                }
            }
            return;
            // break;
        default:
            break;
    }

    std::lock_guard<std::mutex> daemon_lock{g_client->daemo_mutex};
    g_client->daemon_cond_msgid = recv_msg_header->msg_id;
    g_client->daemon_cond_msgdata = resp;
    g_client->daemon_cond_ret = ret;
    g_client->daemon_cond.notify_one();
}

static void on_recv_remote(hio_t *io, void *buf, int readbytes)
{
    uint8_t *buffer = (uint8_t *)buf;
    LOG_PRINT_INFO("on_recv_remote fd[%d], readbytes[%d]", hio_fd(io), readbytes);
    if (readbytes < (int)LOCAL_REGISTRY_HV_MSG_REMOTE_HEADER_SIZE)
    {
        LOG_PRINT_ERROR("not a complete msg");
        return;
    }

    uint32_t real_readbytes = (uint32_t)((uint32_t)readbytes - LOCAL_REGISTRY_HV_MSG_REMOTE_HEADER_SIZE);
    // 使用memcpy避免指针类型转换对齐警告
    uint32_t client_id, msg_seqid, msg_type, service_id, msgdata_len;
    memcpy(&client_id, buffer + 0, sizeof(uint32_t));
    memcpy(&msg_seqid, buffer + 4, sizeof(uint32_t));
    memcpy(&msg_type, buffer + 8, sizeof(uint32_t));
    memcpy(&service_id, buffer + 12, sizeof(uint32_t));
    memcpy(&msgdata_len, buffer + 16, sizeof(uint32_t));

    if (real_readbytes != msgdata_len)
    {
        LOG_PRINT_ERROR("msg_len[%u] != real_readbytes[%u]", msgdata_len, real_readbytes);
        return;
    }

    LOG_PRINT_INFO("on_recv_remote service_id[%u], msg_type[%u], msg_seqid[%d], msg_len[%u]", service_id, msg_type, msg_seqid, msgdata_len);
    if (msg_type != E_IPC_HV_SOA_MSG_TYPE_METHOD_RESPONSE_SYNC)
    {
        ipc_hv_soa_remote_client_data client_data = {};
        client_data.service_id = service_id;
        client_data.msg_type = msg_type;
        client_data.msg_seqid = msg_seqid;
        client_data.client_id = client_id;

        if (msgdata_len > 0)
        {
            client_data.data_len = msgdata_len;
            client_data.data = (uint8_t *)calloc(1, msgdata_len);
            if (nullptr == client_data.data)
            {
                return;
            }
            memcpy(client_data.data, buffer + LOCAL_REGISTRY_HV_MSG_REMOTE_HEADER_SIZE, msgdata_len);
        }
        else
        {
            client_data.data_len = msgdata_len;
            client_data.data = nullptr;
        }
        g_client->msg_handler_queue.Push(client_data);
    }
    else
    {
        void *resp = nullptr;
        if (msgdata_len > 0)
        {
            resp = (void *)calloc(1, msgdata_len);
            if (nullptr == resp)
            {
                LOG_PRINT_ERROR("calloc fail!, errno[%d]", errno);
                return;
            }
            memset(resp, 0x00, msgdata_len);
            memcpy(resp, buffer + LOCAL_REGISTRY_HV_MSG_REMOTE_HEADER_SIZE, msgdata_len);
        }
        ipc_hv_soa_inn_sync_complete(service_id, msg_seqid, &resp, msgdata_len);
    }
}

static void on_write(hio_t *io, const void *buf, int writebytes)
{
    (void)buf;
    if (!hio_write_is_complete(io))
    {
        return;
    }

    LOG_PRINT_DEBUG("on_write fd[%d], writebytes[%d]", hio_fd(io), writebytes);
    if (nullptr != g_client->daemon_io && hio_fd(io) == hio_fd(g_client->daemon_io))
    {
        hio_setcb_read(io, on_recv_daemon);
        hio_set_unpack(io, &g_daemon_unpack_setting);
    }
    else
    {
        hio_setcb_read(io, on_recv_remote);
        hio_set_unpack(io, &g_remote_unpack_setting);
    }

    hio_read(io);
}

static void on_connect(hio_t *io)
{
    LOG_PRINT_DEBUG("on_connect fd[%d]", hio_fd(io));
    if (hio_is_connected(io))
    {
        hio_setcb_write(io, on_write);
        if (nullptr != g_client->daemon_io && hio_fd(io) == hio_fd(g_client->daemon_io))
        {
            if (nullptr == g_client->daemon_io)
            {
                return;
            }
            LOG_PRINT_ERROR("connected with daemon!");
            std::unique_lock daemon_lock(g_client->daemo_mutex);
            g_client->daemon_cond_ret = INT32_MIN;
            g_client->daemon_cond.notify_one();
        }
        else
        {
            uint32_t client_id = *static_cast<uint32_t *>(hevent_userdata(io));
            std::shared_ptr<ipc_hv_soa_remote_client> client = find_remote_client(client_id);
            if (nullptr == client)
            {
                LOG_PRINT_ERROR("invalid client[%d]!", client_id);
            }
            else
            {
                if (nullptr != client->client_send_io && hio_fd(io) == hio_fd(client->client_send_io))
                {
                    LOG_PRINT_ERROR("connected with remote client[%s]!", client->client_name.c_str());
                    std::unique_lock<std::mutex> send_msg_lock(client->send_msg_mutex);
                    client->send_msg_seqid = 0;
                    client->send_msg_list.clear();
                    client->send_msg_cond_ret = INT32_MIN;
                    client->send_msg_cond.notify_one();
                }
            }
        }
    }
    else
    {
        LOG_PRINT_ERROR("peer is closed");
    }
}

static void on_post_event_cb(hevent_t *ev)
{
    hloop_t *loop = ev->loop;
    hio_t *io = (hio_t *)hevent_userdata(ev);
    hio_attach(loop, io);

    hio_setcb_close(io, on_close);
    hio_setcb_read(io, on_recv_remote);
    hio_set_unpack(io, &g_remote_unpack_setting);
    hio_read(io);
}

static void on_accept(hio_t *io)
{
    char localaddrstr[SOCKADDR_STRLEN] = {0};
    char peeraddrstr[SOCKADDR_STRLEN] = {0};
    LOG_PRINT_INFO("on_accept connfd=%d [%s] <= [%s]", hio_fd(io),
                   SOCKADDR_STR(hio_localaddr(io), localaddrstr),
                   SOCKADDR_STR(hio_peeraddr(io), peeraddrstr));

    uint32_t src_id = 0, target_id = 0;
    char src_clientname[LOCAL_REGISTRY_CLIENT_NAME_MAX] = {0};
    int matched = sscanf(peeraddrstr, LOCAL_REGISTEY_SOCKET_FMT2, &src_id, src_clientname, &target_id);
    if (matched == 2)
    {
        LOG_PRINT_DEBUG("src_id[%d], target_id[%d]", src_id, target_id);
        std::shared_ptr<ipc_hv_soa_remote_client> client = nullptr;
        auto it = find_remote_client(src_id);
        if (nullptr == it)
        {
            client = save_remote_client(src_id, src_clientname);
        }
        else
        {
            client = it;
            client->client_status = local_client_status_online;
        }
        hevent_set_userdata(io, &client->client_id);
        client->client_recv_io = io;
    }
    else
    {
        LOG_PRINT_ERROR("sscanf fail!");
    }

    hio_detach(io);

    hloop_t *idle_loop = get_idle_loop();

    hevent_t ev = {};
    ev.loop = idle_loop;
    ev.cb = on_post_event_cb;
    ev.userdata = io;
    hloop_post_event(idle_loop, &ev);
}

static std::shared_ptr<ipc_hv_soa_remote_client> find_remote_client(uint32_t client_id)
{
    std::lock_guard<std::mutex> remote_clients_map_lock(g_remote_clients_map_mutex);
    auto it = g_remote_clients_map.find(client_id);
    if (g_remote_clients_map.end() == it)
    {
        LOG_PRINT_ERROR("client[%d] not find", client_id);
        return nullptr;
    }

    return it->second;
}

static std::shared_ptr<ipc_hv_soa_remote_client> save_remote_client(uint32_t client_id, std::string client_name)
{
    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    std::shared_ptr<ipc_hv_soa_remote_client> client = std::make_shared<ipc_hv_soa_remote_client>();
    client->client_id = client_id;
    client->client_name = client_name;
    char client_localaddr1[LOCAL_REGISTRY_SOCKET_LEN_MAX] = {};
    snprintf(client_localaddr1, sizeof(client_localaddr1), LOCAL_REGISTEY_SOCKET_FMT1, client_id);
    client->client_localaddr1 = client_localaddr1;
    client->client_status = local_client_status_init;
    client->client_send_io = nullptr;
    client->client_recv_io = nullptr;
    client->send_msg_seqid = 0;
    client->send_msg_list.clear();
    client->send_msg_cond_ret = -1;

    if (client_id != g_client->client_id)
    {
        {
            std::lock_guard<std::mutex> remote_clients_map_lock(g_remote_clients_map_mutex);
            g_remote_clients_map.insert({client_id, client});
        }
        ret = connect_with_remote_client(client);
        if (IPC_HV_SOA_RET_SUCCESS != ret)
        {
            LOG_PRINT_ERROR("connect_with_remote_client fail, ret[%d]!", ret);
        }
    }
    else
    {
        client->client_status = local_client_status_online;
        std::lock_guard<std::mutex> remote_clients_map_lock(g_remote_clients_map_mutex);
        g_remote_clients_map.insert({client_id, client});
    }

    return client;
}

static std::shared_ptr<ipc_hv_soa_remote_client> get_remote_client(uint32_t client_id)
{
    int32_t ret = IPC_HV_SOA_RET_SUCCESS;
    std::shared_ptr<ipc_hv_soa_remote_client> client = nullptr;
    auto it = find_remote_client(client_id);
    if (nullptr == it)
    {
        st_get_client_req req = st_get_client_req_init_zero;
        req.client_id = client_id;

        if (g_client->client_status == local_client_status_online)
        {
            std::unique_lock daemon_lock(g_client->daemo_mutex);
            ret = send_msg_to_daemon(g_client->daemon_io, local_registry_msg_get_client, &req, st_get_client_req_fields, st_get_client_req_size);
            if (IPC_HV_SOA_RET_SUCCESS != ret)
            {
                LOG_PRINT_ERROR("send_msg_to_daemon fail, ret[%d]!", ret);
                return nullptr;
            }

            // init
            g_client->daemon_cond_msgid = 0;
            if (nullptr != g_client->daemon_cond_msgdata)
            {
                free(g_client->daemon_cond_msgdata);
                g_client->daemon_cond_msgdata = nullptr;
            }
            g_client->daemon_cond_ret = -1;

            st_get_client_resp *resp = nullptr;
            auto timeout = std::chrono::milliseconds(LOCAL_REGISTRY_COMMUNICATION_TIMEOUT_MS);
            bool ready = g_client->daemon_cond.wait_for(daemon_lock, timeout, []
                                                        {
                                                            return (g_client->daemon_cond_ret != -1);
                                                        });
            if (!ready)
            {
                ret = IPC_HV_SOA_RET_TIMEOUT;
            }
            else
            {
                if (g_client->daemon_cond_ret == 0 && g_client->daemon_cond_msgid == local_registry_msg_get_client)
                {
                    resp = (st_get_client_resp *)g_client->daemon_cond_msgdata;
                    if (resp->has_responser_client)
                    {
                        client = save_remote_client(resp->responser_client.client_id, resp->responser_client.client_name);
                    }
                }
                else
                {
                    LOG_PRINT_ERROR("register client_id fail, invalid daemon_cond_ret[%d], daemon_cond_msgdid[%d]!", g_client->daemon_cond_ret, g_client->daemon_cond_msgid);
                    ret = IPC_HV_SOA_RET_FAIL;
                }
            }

            // clean
            g_client->daemon_cond_msgid = 0;
            if (nullptr != g_client->daemon_cond_msgdata)
            {
                free(g_client->daemon_cond_msgdata);
                g_client->daemon_cond_msgdata = nullptr;
            }
            g_client->daemon_cond_ret = -1;
        }
        else
        {
            LOG_PRINT_ERROR("g_client is offline");
            return nullptr;
        }
    }
    else
    {
        client = it;
    }

    return client;
}

static std::shared_ptr<ipc_hv_soa_service> find_service(uint32_t service_id)
{
    std::lock_guard<std::mutex> services_map_lock(g_services_map_mutex);
    auto it = g_services_map.find(service_id);
    if (g_services_map.end() == it)
    {
        return nullptr;
    }

    return it->second;
}

static std::shared_ptr<ipc_hv_soa_service> save_service(uint32_t service_id, uint32_t service_type, uint32_t service_status, void *service_handler, std::shared_ptr<ipc_hv_soa_remote_client> client)
{
    std::shared_ptr<ipc_hv_soa_service> service_item = nullptr;
    std::lock_guard<std::mutex> services_map_lock(g_services_map_mutex);
    auto it = g_services_map.find(service_id);
    if (nullptr == it)
    {
        service_item = std::make_shared<ipc_hv_soa_service>();
        service_item->service_id = service_id;
        service_item->service_type = service_type;
        if (nullptr == client)
        {
            service_item->service_status = local_service_status_default;
        }
        else
        {
            service_item->service_status = service_status;
        }
        service_item->service_handler = service_handler;
        service_item->provider = nullptr;
        service_item->listeners.clear();
        g_services_map.insert({service_item->service_id, service_item});
    }
    else
    {
        service_item = it->second;
        service_item->service_type = service_type;
        service_item->service_status = service_status;
        if (nullptr != service_handler)
        {
            service_item->service_handler = service_handler;
        }
    }

    if (nullptr != client)
    {
        service_item->provider = client;
    }

    return service_item;
}

static std::shared_ptr<ipc_hv_soa_service> get_service(uint32_t service_id)
{
    int32_t ret = IPC_HV_SOA_RET_SUCCESS;
    std::shared_ptr<ipc_hv_soa_service> service_item = nullptr;

    auto it = find_service(service_id);
    if (nullptr == it ||
        (it->service_type == E_IPC_HV_SOA_SERVICE_TYPE_METHOD && it->service_status == local_service_status_default))
    {
        if (g_client->client_status == local_client_status_online)
        {
            st_get_service_req get_service = st_get_service_req_init_zero;
            get_service.service_id = service_id;
            get_service.has_consumer_client = true;
            get_service.consumer_client.client_id = g_client->client_id;
            get_service.consumer_client.client_pid = g_client->client_pid;
            std::strncpy(get_service.consumer_client.client_name, g_client->client_name.c_str(), sizeof(get_service.consumer_client.client_name));
            get_service.consumer_client.client_status = g_client->client_status;
            std::unique_lock daemon_lock(g_client->daemo_mutex);
            ret = send_msg_to_daemon(g_client->daemon_io, local_registry_msg_get_service, &get_service, st_get_service_req_fields, st_get_service_req_size);
            if (IPC_HV_SOA_RET_SUCCESS != ret)
            {
                LOG_PRINT_ERROR("send_msg_to_daemon fail, ret[%d]!", ret);
                return nullptr;
            }

            // init
            g_client->daemon_cond_msgid = 0;
            g_client->daemon_cond_ret = -1;
            if (nullptr != g_client->daemon_cond_msgdata)
            {
                free(g_client->daemon_cond_msgdata);
                g_client->daemon_cond_msgdata = nullptr;
            }

            st_get_service_resp *resp = nullptr;
            auto timeout = std::chrono::milliseconds(LOCAL_REGISTRY_COMMUNICATION_TIMEOUT_MS);
            bool ready = g_client->daemon_cond.wait_for(daemon_lock, timeout, []
                                                        {
                                                            return g_client->daemon_cond_ret != -1;
                                                        });
            if (!ready)
            {
                LOG_PRINT_ERROR("get service timeout");
                ret = IPC_HV_SOA_RET_TIMEOUT;
            }
            else
            {
                if (g_client->daemon_cond_ret == 0 && g_client->daemon_cond_msgid == local_registry_msg_get_service)
                {
                    resp = (st_get_service_resp *)g_client->daemon_cond_msgdata;
                    LOG_PRINT_DEBUG("get service resp service_id[%u] service_type[%u], service_status[%u]",
                                    resp->service.service_id, resp->service.service_type, resp->service.service_status);
                    std::shared_ptr<ipc_hv_soa_remote_client> client = nullptr;
                    if (resp->has_provider_client)
                    {
                        auto it2 = find_remote_client(resp->provider_client.client_id);
                        if (nullptr == it2)
                        {
                            client = save_remote_client(resp->provider_client.client_id, resp->provider_client.client_name);
                        }
                        else
                        {
                            client = it2;
                        }
                    }
                    service_item = save_service(resp->service.service_id, resp->service.service_type, resp->service.service_status, nullptr, client);
                }
                else
                {
                    LOG_PRINT_ERROR("get service fail, invalid daemon_cond_ret[%d], daemon_cond_msgid[%d]!", g_client->daemon_cond_ret, g_client->daemon_cond_msgid);
                    ret = IPC_HV_SOA_RET_FAIL;
                }

                // clean
                g_client->daemon_cond_msgid = 0;
                g_client->daemon_cond_ret = -1;
                if (nullptr != g_client->daemon_cond_msgdata)
                {
                    free(g_client->daemon_cond_msgdata);
                    g_client->daemon_cond_msgdata = nullptr;
                }
            }
        }
        else
        {
            LOG_PRINT_ERROR("g_client is offline");
            ret = IPC_HV_SOA_RET_FAIL;
        }
    }
    else
    {
        service_item = it;
    }

    return service_item;
}

static std::shared_ptr<ipc_hv_soa_service> get_service_listener(uint32_t service_id)
{
    int32_t ret = IPC_HV_SOA_RET_SUCCESS;
    auto it = get_service(service_id);
    if (nullptr == it)
    {
        LOG_PRINT_ERROR("service[%d] not find", service_id);
        return nullptr;
    }

    if (!it->listeners.empty())
    {
        return it;
    }

    st_get_listener_req req = st_get_listener_req_init_zero;
    req.service_id = service_id;
    std::unique_lock daemon_lock(g_client->daemo_mutex);
    ret = send_msg_to_daemon(g_client->daemon_io, local_registry_msg_get_listener, &req, st_get_listener_req_fields, st_get_listener_req_size);
    if (IPC_HV_SOA_RET_SUCCESS != ret)
    {
        LOG_PRINT_ERROR("send_msg_to_daemon fail, ret[%d]!", ret);
        return nullptr;
    }

    // init
    g_client->daemon_cond_msgid = 0;
    g_client->daemon_cond_ret = -1;
    if (nullptr != g_client->daemon_cond_msgdata)
    {
        free(g_client->daemon_cond_msgdata);
        g_client->daemon_cond_msgdata = nullptr;
    }

    st_get_listener_resp *resp = nullptr;
    auto timeout = std::chrono::milliseconds(LOCAL_REGISTRY_COMMUNICATION_TIMEOUT_MS);
    bool ready = g_client->daemon_cond.wait_for(daemon_lock, timeout, []
                                                {
                                                    return g_client->daemon_cond_ret != -1;
                                                });
    if (!ready)
    {
        LOG_PRINT_ERROR("get listener timeout");
        ret = IPC_HV_SOA_RET_TIMEOUT;
    }
    else
    {
        if (g_client->daemon_cond_ret == 0 && g_client->daemon_cond_msgid == local_registry_msg_get_listener)
        {
            resp = (st_get_listener_resp *)g_client->daemon_cond_msgdata;
            LOG_PRINT_DEBUG("get listener resp listener_clients_count[%d]", resp->listener_clients_count);
            std::shared_ptr<ipc_hv_soa_remote_client> client = nullptr;
            for (uint32_t i = 0; i < resp->listener_clients_count; i++)
            {
                LOG_PRINT_DEBUG("get listener resp listener_client_id[%d] listener_client_name[%s]", resp->listener_clients[i].client_id, resp->listener_clients[i].client_name);
                auto it2 = find_remote_client(resp->listener_clients[i].client_id);
                if (nullptr == it2)
                {
                    client = save_remote_client(resp->listener_clients[i].client_id, resp->listener_clients[i].client_name);
                }
                else
                {
                    client = it2;
                }
                it->listeners.insert({client->client_id, client});
            }
        }
        else
        {
            LOG_PRINT_ERROR("get listener fail, invalid daemon_cond_ret[%d], daemon_cond_msgid[%d]!", g_client->daemon_cond_ret, g_client->daemon_cond_msgid);
            ret = IPC_HV_SOA_RET_FAIL;
        }

        // clean
        g_client->daemon_cond_msgid = 0;
        g_client->daemon_cond_ret = -1;
        if (nullptr != g_client->daemon_cond_msgdata)
        {
            free(g_client->daemon_cond_msgdata);
            g_client->daemon_cond_msgdata = nullptr;
        }
    }

    return it;
}

static int32_t send_msg_to_daemon(hio_t *daemon_io, uint32_t msg_id, const void *msgdata, const pb_msgdesc_t *fileds, uint32_t field_size)
{
    int32_t ret = 0;
    if (nullptr == daemon_io)
    {
        LOG_PRINT_ERROR("daemon_io is nullptr");
        return -1;
    }

    LOG_PRINT_DEBUG("send msg_id[%d] to daemon fd[%d]", msg_id, hio_fd(daemon_io));
    std::vector<uint8_t> buffer(LOCAL_REGISTRY_HV_MSG_HEADER_SIZE + field_size, 0);
    pb_ostream_t stream = pb_ostream_from_buffer(buffer.data() + LOCAL_REGISTRY_HV_MSG_HEADER_SIZE, field_size);
    bool status = pb_encode(&stream, fileds, msgdata);
    if (!status)
    {
        LOG_PRINT_ERROR("pb_encode msg_id[%d] fail, error(%s)", msg_id, PB_GET_ERROR(&stream));
        ret = -1;
    }
    else
    {
        st_local_msg_header send_msg_header = {};
        send_msg_header.msg_id = msg_id;
        send_msg_header.msg_len = (uint32_t)stream.bytes_written;
        memcpy(buffer.data(), &send_msg_header, sizeof(send_msg_header));
        if (hio_is_opened(daemon_io))
        {
            ret = hio_write(daemon_io, buffer.data(), LOCAL_REGISTRY_HV_MSG_HEADER_SIZE + stream.bytes_written);
            if (ret < 0)
            {
                LOG_PRINT_ERROR("hio_write fail, ret[%d], error[%d]", ret, hio_error(daemon_io));
            }
            else
            {
                ret = IPC_HV_SOA_RET_SUCCESS;
            }
        }
        else
        {
            LOG_PRINT_ERROR("hio_write fail, fd[%d] disconnect", hio_fd(daemon_io));
            ret = IPC_HV_SOA_RET_FAIL;
        }
    }

    return ret;
}

static int32_t connect_with_daemon()
{
    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    g_client->daemon_io = hio_create_socket(g_client->daemon_loop, LOCAL_REGISTRY_SOCKET_PATH, -1, HIO_TYPE_SOCK_STREAM, HIO_CLIENT_SIDE);
    if (nullptr == g_client->daemon_io)
    {
        LOG_PRINT_ERROR("hio_create_socket fail!");
        return IPC_HV_SOA_RET_FAIL;
    }

    unlink(g_client->client_localaddr.c_str());
    struct sockaddr_un client_addr = {};
    client_addr.sun_family = AF_UNIX;
    strncpy(client_addr.sun_path, g_client->client_localaddr.c_str(), sizeof(client_addr.sun_path));
    if (bind(hio_fd(g_client->daemon_io), (struct sockaddr *)&client_addr, (socklen_t)SUN_LEN(&client_addr)))
    {
        LOG_PRINT_ERROR("bind %s fail, errno[%d](%s)!", client_addr.sun_path, errno, strerror(errno));
        return IPC_HV_SOA_RET_FAIL;
    }

    std::unique_lock daemon_lock(g_client->daemo_mutex);

    // init
    g_client->daemon_cond_msgid = 0;
    if (nullptr != g_client->daemon_cond_msgdata)
    {
        free(g_client->daemon_cond_msgdata);
        g_client->daemon_cond_msgdata = nullptr;
    }
    g_client->daemon_cond_ret = -1;

    hio_setcb_close(g_client->daemon_io, on_close);
    hio_setcb_connect(g_client->daemon_io, on_connect);
    hio_set_connect_timeout(g_client->daemon_io, 1000);
    if (0 != hio_connect(g_client->daemon_io))
    {
        LOG_PRINT_ERROR("hio_connect fail, errno[%d]!", hio_error(g_client->daemon_io));
        return IPC_HV_SOA_RET_FAIL;
    }

    auto timeout = std::chrono::milliseconds(LOCAL_REGISTRY_COMMUNICATION_TIMEOUT_MS);
    bool ready = g_client->daemon_cond.wait_for(daemon_lock, timeout, []
                                                {
                                                    return (g_client->daemon_cond_ret != -1);
                                                });
    if (!ready)
    {
        ret = IPC_HV_SOA_RET_TIMEOUT;
    }
    else
    {
        if (g_client->daemon_cond_ret != INT32_MIN)
        {
            LOG_PRINT_ERROR("connect with daemon fail, invalid daemon_cond_ret[%d]!", g_client->daemon_cond_ret);
            ret = IPC_HV_SOA_RET_FAIL;
        }
        else
        {
            g_client->client_status = local_client_status_online;
        }

        // clean
        g_client->daemon_cond_msgid = 0;
        if (nullptr != g_client->daemon_cond_msgdata)
        {
            free(g_client->daemon_cond_msgdata);
            g_client->daemon_cond_msgdata = nullptr;
        }
        g_client->daemon_cond_ret = -1;
    }

    return ret;
}

static int32_t connect_with_remote_client(std::shared_ptr<ipc_hv_soa_remote_client> client)
{
    int32_t ret = IPC_HV_SOA_RET_SUCCESS;
    hloop_t *idle_loop = get_idle_loop();

    if (nullptr != client->client_send_io)
    {
        hio_close(client->client_send_io);
        return IPC_HV_SOA_RET_FAIL;
    }

    client->client_send_io = hio_create_socket(idle_loop, client->client_localaddr1.c_str(), -1, HIO_TYPE_SOCK_STREAM, HIO_CLIENT_SIDE);
    if (nullptr == client->client_send_io)
    {
        LOG_PRINT_ERROR("hio_create_socket fail!");
        return IPC_HV_SOA_RET_FAIL;
    }

    struct sockaddr_un client_addr = {};
    client_addr.sun_family = AF_UNIX;
    snprintf(client_addr.sun_path, sizeof(client_addr.sun_path), LOCAL_REGISTEY_SOCKET_FMT2, g_client->client_id, g_client->client_name.c_str(), client->client_id);
    unlink(client_addr.sun_path);
    if (bind(hio_fd(client->client_send_io), (struct sockaddr *)&client_addr, (socklen_t)SUN_LEN(&client_addr)))
    {
        LOG_PRINT_ERROR("bind %s fail, errno[%d](%s)", client_addr.sun_path, errno, strerror(errno));
        return IPC_HV_SOA_RET_FAIL;
    }

    std::unique_lock send_msg_lock(client->send_msg_mutex);

    // init
    client->send_msg_seqid = 0;
    client->send_msg_list.clear();
    client->send_msg_cond_ret = -1;

    hio_setcb_close(client->client_send_io, on_close);
    hio_setcb_connect(client->client_send_io, on_connect);
    hio_set_connect_timeout(client->client_send_io, 1000);
    hevent_set_userdata(client->client_send_io, &client->client_id);
    if (0 != hio_connect(client->client_send_io))
    {
        LOG_PRINT_ERROR("hio_connect fail, errno[%d]!", hio_error(client->client_send_io));
        return IPC_HV_SOA_RET_FAIL;
    }

    auto timeout = std::chrono::milliseconds(LOCAL_REGISTRY_COMMUNICATION_TIMEOUT_MS);
    bool ready = client->send_msg_cond.wait_for(send_msg_lock, timeout, [&client]
                                                {
                                                    return (client->send_msg_cond_ret != -1);
                                                });
    if (!ready)
    {
        ret = IPC_HV_SOA_RET_TIMEOUT;
    }
    else
    {
        if (client->send_msg_cond_ret != INT32_MIN)
        {
            LOG_PRINT_ERROR("connect with daemon fail, invalid send_msg_cond_ret[%d]!", client->send_msg_cond_ret);
            ret = IPC_HV_SOA_RET_FAIL;
        }
        else
        {
            LOG_PRINT_ERROR("connected with remote client[%s], it be online", client->client_name.c_str());
            client->client_status = local_client_status_online;
        }
    }

    // clean
    client->send_msg_cond_ret = -1;

    return ret;
}

static int32_t send_msg_to_client(std::shared_ptr<ipc_hv_soa_remote_client> dest, uint32_t client_id, uint32_t msg_seqid, uint32_t msg_type, uint32_t service_id, uint32_t msgdata_len, const void *msgdata)
{
    if (nullptr == dest)
    {
        LOG_PRINT_ERROR("invalid params!");
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    if (dest->client_status != local_client_status_online)
    {
        ret = connect_with_remote_client(dest);
    }
    else
    {
        if (nullptr == dest->client_send_io || !hio_is_opened(dest->client_send_io))
        {
            LOG_PRINT_ERROR("remote client[%d] is offline!", dest->client_id);
            return IPC_HV_SOA_RET_FAIL;
        }
    }

    if (IPC_HV_SOA_RET_SUCCESS == ret)
    {
        std::lock_guard<std::mutex> send_msg_lock(dest->send_msg_mutex);
        LOG_PRINT_DEBUG("client[%d] send msg_id[%d] msg_type[%d] msg_seqid[%d] to client[%d]", client_id, service_id, msg_type, msg_seqid, dest->client_id);
        std::vector<uint8_t> buffer(sizeof(uint32_t) * 5 + msgdata_len);
        memcpy(buffer.data() + 0 * sizeof(uint32_t), &client_id, sizeof(uint32_t));
        memcpy(buffer.data() + 1 * sizeof(uint32_t), &msg_seqid, sizeof(uint32_t));
        memcpy(buffer.data() + 2 * sizeof(uint32_t), &msg_type, sizeof(uint32_t));
        memcpy(buffer.data() + 3 * sizeof(uint32_t), &service_id, sizeof(uint32_t));
        memcpy(buffer.data() + 4 * sizeof(uint32_t), &msgdata_len, sizeof(uint32_t));
        if (nullptr != msgdata)
        {
            memcpy(buffer.data() + sizeof(uint32_t) * 5, msgdata, msgdata_len);
        }

        ret = hio_write(dest->client_send_io, buffer.data(), buffer.size());
        if (ret < 0)
        {
            LOG_PRINT_ERROR("hio_write fail, ret[%d], error[%d]", ret, hio_error(dest->client_send_io));
            return IPC_HV_SOA_RET_FAIL;
        }
        else
        {
            ret = IPC_HV_SOA_RET_SUCCESS;
        }
    }
    else
    {
        LOG_PRINT_ERROR("connect with client[%d] fail!", dest->client_id);
    }

    return ret;
}

static int32_t send_msg_to_client_sync(std::shared_ptr<ipc_hv_soa_remote_client> dest, uint32_t client_id, uint32_t msg_seqid, uint32_t service_id, uint32_t msgdata_len, const void *msgdata, void *method_resp_data, uint32_t *method_resp_data_len, uint32_t timeout_ms)
{
    if (nullptr == dest)
    {
        LOG_PRINT_ERROR("invalid params!");
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    if (dest->client_status != local_client_status_online)
    {
        ret = connect_with_remote_client(dest);
    }
    else
    {
        if (nullptr == dest->client_send_io || !hio_is_opened(dest->client_send_io))
        {
            ret = connect_with_remote_client(dest);
        }
    }

    if (IPC_HV_SOA_RET_SUCCESS == ret)
    {
        std::unique_lock<std::mutex> send_msg_lock(dest->send_msg_mutex);
        ipc_hv_soa_remote_client_data expected_resp_data = {};
        ipc_hv_soa_remote_client_data *p_resp_data = nullptr;

        LOG_PRINT_DEBUG("client[%d] send msg_id[%d] msg_type[%d] msg_seqid[%d] to client[%d]", client_id, service_id, E_IPC_HV_SOA_MSG_TYPE_METHOD_REQUEST_SYNC, msg_seqid, dest->client_id);
        expected_resp_data.service_id = service_id;
        expected_resp_data.msg_type = E_IPC_HV_SOA_MSG_TYPE_METHOD_RESPONSE_SYNC;
        expected_resp_data.msg_seqid = msg_seqid;
        expected_resp_data.client_id = 0;
        expected_resp_data.is_complete = false;
        expected_resp_data.data_len = 0;
        expected_resp_data.data = nullptr;
        dest->send_msg_list.emplace_back(expected_resp_data);

        std::vector<uint8_t> buffer(sizeof(uint32_t) * 5 + msgdata_len);
        uint32_t msg_type = E_IPC_HV_SOA_MSG_TYPE_METHOD_REQUEST_SYNC;
        memcpy(buffer.data() + 0 * sizeof(uint32_t), &client_id, sizeof(uint32_t));
        memcpy(buffer.data() + 1 * sizeof(uint32_t), &msg_seqid, sizeof(uint32_t));
        memcpy(buffer.data() + 2 * sizeof(uint32_t), &msg_type, sizeof(uint32_t));
        memcpy(buffer.data() + 3 * sizeof(uint32_t), &service_id, sizeof(uint32_t));
        memcpy(buffer.data() + 4 * sizeof(uint32_t), &msgdata_len, sizeof(uint32_t));
        if (nullptr != msgdata)
        {
            memcpy(buffer.data() + sizeof(uint32_t) * 5, msgdata, msgdata_len);
        }

        ret = hio_write(dest->client_send_io, buffer.data(), buffer.size());
        if (ret < 0)
        {
            LOG_PRINT_ERROR("hio_write fail, ret[%d], error[%d]", ret, hio_error(dest->client_send_io));
            ret = IPC_HV_SOA_RET_FAIL;
        }
        else
        {
            // init
            auto timeout = std::chrono::milliseconds(timeout_ms);
            dest->send_msg_cond_ret = -1;
            do
            {
                bool ready = dest->send_msg_cond.wait_for(send_msg_lock, timeout, [&dest]
                                                          {
                                                              return (dest->send_msg_cond_ret != -1);
                                                          });
                if (!ready)
                {
                    ret = IPC_HV_SOA_RET_TIMEOUT;
                    break;
                }

                for (auto &resp_data : dest->send_msg_list)
                {
                    if (resp_data.service_id == service_id && resp_data.msg_seqid == msg_seqid)
                    {
                        p_resp_data = &resp_data;
                        break;
                    }
                }

                if (nullptr == p_resp_data)
                {
                    ret = IPC_HV_SOA_RET_FAIL;
                    break;
                }

                if (p_resp_data->is_complete)
                {
                    ret = IPC_HV_SOA_RET_SUCCESS;
                    break;
                }
            } while (1);

            if (IPC_HV_SOA_RET_SUCCESS == ret)
            {
                if (nullptr != method_resp_data && nullptr != method_resp_data_len)
                {
                    if (*method_resp_data_len < p_resp_data->data_len)
                    {
                        ret = IPC_HV_SOA_RET_ERR_ARG;
                    }
                    else
                    {
                        if (nullptr != p_resp_data->data && p_resp_data->data_len > 0)
                        {
                            memcpy(method_resp_data, p_resp_data->data, p_resp_data->data_len);
                        }
                    }
                }

                if (nullptr != method_resp_data_len)
                {
                    *method_resp_data_len = p_resp_data->data_len;
                }

                if (nullptr != p_resp_data->data)
                {
                    free(p_resp_data->data);
                }
            }
            else
            {
                LOG_PRINT_ERROR("ipc_hv_soa_method_sync[%d] fail, ret[%d]", service_id, ret);
            }
        }
        dest->send_msg_list.remove_if([&p_resp_data](const ipc_hv_soa_remote_client_data &item)
                                      {
                                          return item.service_id == p_resp_data->service_id &&
                                              item.msg_seqid == p_resp_data->msg_seqid;
                                      });
    }
    else
    {
        LOG_PRINT_ERROR("connect with client[%d] fail!", dest->client_id);
    }

    return ret;
}

static int32_t register_client_req()
{
    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    st_register_client_req req = st_register_client_req_init_zero;
    req.client_pid = g_client->client_pid;
    std::strncpy(req.client_name, g_client->client_name.c_str(), sizeof(req.client_name));
    LOG_PRINT_DEBUG("register client id req client_pid[%d], client_name[%s]", req.client_pid, req.client_name);

    if (g_client->client_status == local_client_status_online)
    {
        std::unique_lock daemon_lock(g_client->daemo_mutex);
        ret = send_msg_to_daemon(g_client->daemon_io, local_registry_msg_register_client, &req, st_register_client_req_fields, st_register_client_req_size);
        if (IPC_HV_SOA_RET_SUCCESS != ret)
        {
            LOG_PRINT_ERROR("send_msg_to_daemon fail, ret[%d]!", ret);
            return ret;
        }

        // init
        g_client->daemon_cond_msgid = 0;
        if (nullptr != g_client->daemon_cond_msgdata)
        {
            free(g_client->daemon_cond_msgdata);
            g_client->daemon_cond_msgdata = nullptr;
        }
        g_client->daemon_cond_ret = -1;

        st_register_client_resp *resp = nullptr;
        auto timeout = std::chrono::milliseconds(LOCAL_REGISTRY_COMMUNICATION_TIMEOUT_MS);
        bool ready = g_client->daemon_cond.wait_for(daemon_lock, timeout, []
                                                    {
                                                        return (g_client->daemon_cond_ret != -1);
                                                    });
        if (!ready)
        {
            ret = IPC_HV_SOA_RET_TIMEOUT;
        }
        else
        {
            if (g_client->daemon_cond_ret == 0 && g_client->daemon_cond_msgid == local_registry_msg_register_client)
            {
                resp = (st_register_client_resp *)g_client->daemon_cond_msgdata;
                if (resp->client_pid != g_client->client_pid)
                {
                    LOG_PRINT_ERROR("client_pid[%d] != resp->client_pid[%d]", g_client->client_pid, resp->client_pid);
                    ret = IPC_HV_SOA_RET_FAIL;
                }
                else
                {
                    g_client->client_id = resp->client_id;
                    ret = IPC_HV_SOA_RET_SUCCESS;
                }
            }
            else
            {
                LOG_PRINT_ERROR("register client_id fail, invalid daemon_cond_ret[%d], daemon_cond_msgdid[%d]!", g_client->daemon_cond_ret, g_client->daemon_cond_msgid);
                ret = IPC_HV_SOA_RET_FAIL;
            }

            // clean
            g_client->daemon_cond_msgid = 0;
            if (nullptr != g_client->daemon_cond_msgdata)
            {
                free(g_client->daemon_cond_msgdata);
                g_client->daemon_cond_msgdata = nullptr;
            }
            g_client->daemon_cond_ret = -1;
        }
    }
    else
    {
        LOG_PRINT_ERROR("g_client is offline");
        return IPC_HV_SOA_RET_FAIL;
    }

    return ret;
}

int32_t ipc_hv_soa_init(uint32_t *client_id)
{
    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    if (nullptr == client_id)
    {
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    if (0 != access(LOCAL_REGISTRY_SOCKET_PATH, F_OK))
    {
        LOG_PRINT_ERROR("local_registry is not running!");
        return IPC_HV_SOA_RET_FAIL;
    }

    if (g_init_flag)
    {
        LOG_PRINT_DEBUG("ipc hv soa already inited!");
        return IPC_HV_SOA_RET_SUCCESS;
    }

    g_client = std::make_shared<ipc_hv_soa_client>();

    g_daemon_unpack_setting.mode = UNPACK_BY_LENGTH_FIELD;
    g_daemon_unpack_setting.package_max_length = DEFAULT_PACKAGE_MAX_LENGTH;
    g_daemon_unpack_setting.body_offset = LOCAL_REGISTRY_HV_MSG_HEADER_SIZE;
    g_daemon_unpack_setting.length_field_offset = sizeof(uint32_t);
    g_daemon_unpack_setting.length_field_bytes = sizeof(uint32_t);
    g_daemon_unpack_setting.length_adjustment = 0;
    g_daemon_unpack_setting.length_field_coding = ENCODE_BY_LITTEL_ENDIAN;

    g_remote_unpack_setting.mode = UNPACK_BY_LENGTH_FIELD;
    g_remote_unpack_setting.package_max_length = DEFAULT_PACKAGE_MAX_LENGTH;
    g_remote_unpack_setting.body_offset = LOCAL_REGISTRY_HV_MSG_REMOTE_HEADER_SIZE;
    g_remote_unpack_setting.length_field_offset = sizeof(uint32_t) * 4;
    g_remote_unpack_setting.length_field_bytes = sizeof(uint32_t);
    g_remote_unpack_setting.length_adjustment = 0;
    g_remote_unpack_setting.length_field_coding = ENCODE_BY_LITTEL_ENDIAN;

    {
        std::lock_guard<std::mutex> remote_clients_map_lock(g_remote_clients_map_mutex);
        g_remote_clients_map.clear();
    }

    {
        std::lock_guard<std::mutex> services_map_lock(g_services_map_mutex);
        g_services_map.clear();
    }

    g_client->daemon_loop = hloop_new(HLOOP_FLAG_AUTO_FREE);
    if (nullptr == g_client->daemon_loop)
    {
        LOG_PRINT_ERROR("hloop_new fail!");
        return IPC_HV_SOA_RET_FAIL;
    }
    g_client->daemon_loop_thread = std::thread([loop = g_client->daemon_loop]()
                                               {
                                                   LOG_PRINT_DEBUG("main engine[%s]-pid[%ld]-tid[%ld]", hio_engine(), hloop_pid(loop), hloop_tid(loop));
                                                   hloop_run(loop);
                                               });

    g_client->worker_loops = (hloop_t **)calloc(LOCAL_REGISTRY_HV_LOOP_NUM_MAX, sizeof(hloop_t *));
    if (nullptr == g_client->worker_loops)
    {
        LOG_PRINT_ERROR("calloc fail, errno[%d](%s)", errno, strerror(errno));
        return IPC_HV_SOA_RET_FAIL;
    }

    for (size_t i = 0; i < LOCAL_REGISTRY_HV_LOOP_NUM_MAX; ++i)
    {
        g_client->worker_loops[i] = hloop_new(HLOOP_FLAG_AUTO_FREE);
        if (nullptr == g_client->worker_loops[i])
        {
            LOG_PRINT_ERROR("hloop_new fail!");
            return IPC_HV_SOA_RET_FAIL;
        }
        g_client->worker_threads.emplace_back([loop = g_client->worker_loops[i]]()
                                              {
                                                  LOG_PRINT_DEBUG("worker engine[%s]-pid[%ld]-tid[%ld]", hio_engine(), hloop_pid(loop), hloop_tid(loop));
                                                  hloop_run(loop);
                                              });
    }

    g_client->msg_handler_thread = std::thread([]()
                                               {
                                                   while (1)
                                                   {
                                                       ipc_hv_soa_remote_client_data recv_data = {};
                                                       if (!g_client->msg_handler_queue.Pop(recv_data))
                                                       {
                                                           break;
                                                       }

                                                       ipc_hv_soa_msg_handle_t handle = {};
                                                       handle.client_id = recv_data.client_id;
                                                       handle.msg_type = recv_data.msg_type;
                                                       handle.msg_seqid = recv_data.msg_seqid;

                                                       switch (recv_data.msg_type)
                                                       {
                                                           case E_IPC_HV_SOA_MSG_TYPE_METHOD_NOTIFY:
                                                           case E_IPC_HV_SOA_MSG_TYPE_METHOD_REQUEST_SYNC:
                                                           case E_IPC_HV_SOA_MSG_TYPE_METHOD_REQUEST_ASYNC:
                                                               {
                                                                   auto it = find_service(recv_data.service_id);
                                                                   if (it == nullptr)
                                                                   {
                                                                       LOG_PRINT_ERROR("not find service_id[%u]", recv_data.service_id);
                                                                       break;
                                                                   }

                                                                   PF_IPC_HV_SOA_METHOD_PROVIDER_CB cb = (PF_IPC_HV_SOA_METHOD_PROVIDER_CB)it->service_handler;
                                                                   if (nullptr != cb)
                                                                   {
                                                                       cb(handle, recv_data.service_id, recv_data.data, recv_data.data_len);
                                                                   }
                                                               }
                                                               break;
                                                           case E_IPC_HV_SOA_MSG_TYPE_METHOD_RESPONSE_SYNC:
                                                               break;
                                                           case E_IPC_HV_SOA_MSG_TYPE_METHOD_RESPONSE_ASYNC:
                                                               {
                                                                   auto it = find_service(recv_data.service_id);
                                                                   if (it == nullptr)
                                                                   {
                                                                       LOG_PRINT_ERROR("not find service_id[%u]", recv_data.service_id);
                                                                       break;
                                                                   }
                                                                   PF_IPC_HV_SOA_METHOD_ASYNC_CB cb = (PF_IPC_HV_SOA_METHOD_ASYNC_CB)it->service_handler;
                                                                   if (nullptr != cb)
                                                                   {
                                                                       cb(recv_data.service_id, recv_data.data, recv_data.data_len);
                                                                   }
                                                               }
                                                               break;
                                                           case E_IPC_HV_SOA_MSG_TYPE_EVENT:
                                                               {
                                                                   auto it = find_service(recv_data.service_id);
                                                                   if (it == nullptr)
                                                                   {
                                                                       LOG_PRINT_ERROR("not find service_id[%u]", recv_data.service_id);
                                                                       break;
                                                                   }

                                                                   PF_IPC_HV_SOA_EVENT_LISTENER_CB cb = (PF_IPC_HV_SOA_EVENT_LISTENER_CB)it->service_handler;
                                                                   if (nullptr != cb)
                                                                   {
                                                                       cb(recv_data.service_id, recv_data.data, recv_data.data_len);
                                                                   }
                                                               }
                                                               break;
                                                           default:
                                                               LOG_PRINT_ERROR("invalid msg_type[%u]", recv_data.msg_type);
                                                               break;
                                                       }

                                                       if (recv_data.data_len > 0 && nullptr != recv_data.data)
                                                       {
                                                           free(recv_data.data);
                                                       }
                                                   }
                                               });
    usleep(100 * 1000); // to wait hloop_run is really started

    {
        g_client->client_pid = getpid();
        char client_name[LOCAL_REGISTRY_CLIENT_NAME_MAX] = {};
        if (0 != get_exec_name(g_client->client_pid, client_name, sizeof(client_name)))
        {
            LOG_PRINT_ERROR("get_exec_name fail!");
            return IPC_HV_SOA_RET_FAIL;
        }
        g_client->client_name = client_name;

        char client_localaddr[LOCAL_REGISTRY_SOCKET_LEN_MAX] = {};
        snprintf(client_localaddr, sizeof(client_localaddr), LOCAL_REGISTEY_SOCKET_FMT, g_client->client_pid, client_name);
        g_client->client_localaddr = client_localaddr;

        // get client id from daemon
        ret = connect_with_daemon();
        if (IPC_HV_SOA_RET_SUCCESS != ret)
        {
            LOG_PRINT_ERROR("connect_with_daemon fail, ret[%d]!", ret);
            return IPC_HV_SOA_RET_FAIL;
        }

        ret = register_client_req();
        if (IPC_HV_SOA_RET_SUCCESS != ret)
        {
            LOG_PRINT_ERROR("register_client_req fail, ret[%d]!", ret);
            return IPC_HV_SOA_RET_FAIL;
        }
    }

    {
        char client_localaddr1[LOCAL_REGISTRY_SOCKET_LEN_MAX] = {};
        snprintf(client_localaddr1, sizeof(client_localaddr1), LOCAL_REGISTEY_SOCKET_FMT1, g_client->client_id);
        g_client->client_localaddr1 = client_localaddr1;

        // listen remote clients
        unlink(client_localaddr1);
        g_client->listen_io = hio_create_socket(g_client->daemon_loop, client_localaddr1, -1, HIO_TYPE_SOCK_STREAM, HIO_SERVER_SIDE);
        if (nullptr == g_client->listen_io)
        {
            LOG_PRINT_ERROR("hio_create_socket fail!");
            return IPC_HV_SOA_RET_FAIL;
        }
        hio_setcb_close(g_client->listen_io, on_close);
        hio_setcb_accept(g_client->listen_io, on_accept);
        hio_accept(g_client->listen_io);
    }

    auto it = save_remote_client(g_client->client_id, g_client->client_name);
    *client_id = it->client_id;

    g_init_flag = true;
    ret = IPC_HV_SOA_RET_SUCCESS;

    return ret;
}

int32_t ipc_hv_soa_destroy(uint32_t client_id)
{
    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    LOG_PRINT_DEBUG("ipv hv soa destroy");
    if (g_client && g_client->client_id == client_id)
    {
        if (g_client->daemon_io)
        {
            hio_close(g_client->daemon_io);
        }

        if (g_client->listen_io)
        {
            hio_close(g_client->listen_io);
        }

        hloop_stop(g_client->daemon_loop);
        if (g_client->daemon_loop_thread.joinable())
        {
            g_client->daemon_loop_thread.join();
        }

        for (size_t i = 0; i < LOCAL_REGISTRY_HV_LOOP_NUM_MAX; ++i)
        {
            if (g_client->worker_loops[i])
            {
                hloop_stop(g_client->worker_loops[i]);
            }
        }

        for (auto &thread : g_client->worker_threads)
        {
            if (thread.joinable())
            {
                thread.join();
            }
        }
        free(g_client->worker_loops);
        g_client->worker_loops = nullptr;
    }

    std::lock_guard<std::mutex> remote_clients_map_lock(g_remote_clients_map_mutex);
    for (auto &ptr : g_remote_clients_map)
    {
        if (ptr.second->client_send_io)
        {
            hio_close(ptr.second->client_send_io);
        }

        if (ptr.second->client_recv_io)
        {
            hio_close(ptr.second->client_recv_io);
        }

        ptr.second->send_msg_list.clear();
    }
    g_remote_clients_map.clear();

    g_client->msg_handler_queue.Clear();
    g_client->msg_handler_queue.Quit();
    if (g_client->msg_handler_thread.joinable())
    {
        g_client->msg_handler_thread.join();
    }

    return ret;
}

int32_t ipc_hv_soa_provider_service_offer(ipc_hv_soa_provider_service_t *provider_services, uint32_t provider_services_size)
{
    if (nullptr == g_client || nullptr == provider_services || 0 == provider_services_size)
    {
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    if (provider_services_size >= LOCAL_REGISTRY_CLIENT_PROVIDE_SERVICES_ONCE_COUNT_MAX)
    {
        LOG_PRINT_ERROR("one clinet supporet services limit[%u]", LOCAL_REGISTRY_CLIENT_PROVIDE_SERVICES_ONCE_COUNT_MAX);
        return IPC_HV_SOA_RET_ERR_LIMIT;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;
    auto it = find_remote_client(g_client->client_id);
    if (nullptr == it)
    {
        LOG_PRINT_ERROR("client[%d] not find", g_client->client_id);
        return IPC_HV_SOA_RET_FAIL;
    }

    st_register_service register_service = st_register_service_init_zero;
    for (size_t i = 0; i < provider_services_size; ++i)
    {
        register_service.produce_services[i].service_id = provider_services[i].service_id;
        register_service.produce_services[i].service_type = provider_services[i].service_type;
        register_service.produce_services[i].service_status = local_service_status_available;
        save_service(provider_services[i].service_id, provider_services[i].service_type, register_service.produce_services[i].service_status, provider_services[i].service_handler, it);
    }

    register_service.produce_services_count = (pb_size_t)provider_services_size;
    register_service.has_provider_client = true;
    register_service.provider_client.client_id = g_client->client_id;
    register_service.provider_client.client_pid = g_client->client_pid;
    std::strncpy(register_service.provider_client.client_name, g_client->client_name.c_str(), sizeof(register_service.provider_client.client_name));
    register_service.provider_client.client_status = g_client->client_status;
    register_service.reg = true;

    if (g_client->client_status == local_client_status_online)
    {
        std::unique_lock daemon_lock(g_client->daemo_mutex);
        ret = send_msg_to_daemon(g_client->daemon_io, local_registry_msg_register_service, &register_service, st_register_service_fields, st_register_service_size);
        if (IPC_HV_SOA_RET_SUCCESS != ret)
        {
            LOG_PRINT_ERROR("send_msg_to_daemon fail, ret[%d]!", ret);
        }
    }
    else
    {
        LOG_PRINT_ERROR("g_client is offline");
        ret = IPC_HV_SOA_RET_FAIL;
    }

    return ret;
}

int32_t ipc_hv_soa_provider_service_revoke(uint32_t *provider_services, uint32_t provider_services_size)
{
    if (nullptr == g_client || nullptr == provider_services || 0 == provider_services_size)
    {
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    if (provider_services_size >= LOCAL_REGISTRY_CLIENT_PROVIDE_SERVICES_ONCE_COUNT_MAX)
    {
        LOG_PRINT_ERROR("one clinet supporet services limit[%u]", LOCAL_REGISTRY_CLIENT_PROVIDE_SERVICES_ONCE_COUNT_MAX);
        return IPC_HV_SOA_RET_ERR_LIMIT;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    st_register_service unregister_service = st_register_service_init_zero;
    {
        for (size_t i = 0; i < provider_services_size; ++i)
        {
            auto it = find_service(provider_services[i]);
            if (nullptr == it)
            {
                LOG_PRINT_ERROR("service[%d] not find", provider_services[i]);
                return IPC_HV_SOA_RET_FAIL;
            }

            unregister_service.produce_services[i].service_id = it->service_id;
            unregister_service.produce_services[i].service_type = it->service_type;
            unregister_service.produce_services[i].service_status = local_service_status_unavailable;
            it->service_status = local_service_status_unavailable;
        }
        unregister_service.produce_services_count = (pb_size_t)provider_services_size;
        unregister_service.has_provider_client = true;
        unregister_service.provider_client.client_id = g_client->client_id;
        unregister_service.provider_client.client_pid = g_client->client_pid;
        std::strncpy(unregister_service.provider_client.client_name, g_client->client_name.c_str(), sizeof(unregister_service.provider_client.client_name));
        unregister_service.provider_client.client_status = g_client->client_status;
    }
    unregister_service.reg = false;

    if (g_client->client_status == local_client_status_online)
    {
        std::unique_lock daemon_lock(g_client->daemo_mutex);
        ret = send_msg_to_daemon(g_client->daemon_io, local_registry_msg_register_service, &unregister_service, st_register_service_fields, st_register_service_size);
        if (IPC_HV_SOA_RET_SUCCESS != ret)
        {
            LOG_PRINT_ERROR("send_msg_to_daemon fail, ret[%d]!", ret);
        }
    }
    else
    {
        LOG_PRINT_ERROR("g_client is offline");
        ret = IPC_HV_SOA_RET_FAIL;
    }

    return ret;
}

int32_t ipc_hv_soa_provider_service_set_status(uint32_t *provider_services, uint32_t provider_services_size, uint32_t provider_services_status)
{
    if (nullptr == g_client || nullptr == provider_services || 0 == provider_services_size)
    {
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    if (provider_services_size > LOCAL_REGISTRY_CLIENT_PROVIDE_SERVICES_ONCE_COUNT_MAX)
    {
        return IPC_HV_SOA_RET_ERR_LIMIT;
    }

    st_service_set_status set_status = st_service_set_status_init_zero;
    set_status.services_count = (pb_size_t)provider_services_size;
    for (size_t i = 0; i < provider_services_size; ++i)
    {
        set_status.services[i].service_id = provider_services[i];
        // service_type not used
        set_status.services[i].service_status = provider_services_status;
    }
    set_status.has_provider_client = true;
    set_status.provider_client.client_id = g_client->client_id;
    set_status.provider_client.client_pid = g_client->client_pid;
    std::strncpy(set_status.provider_client.client_name, g_client->client_name.c_str(), sizeof(set_status.provider_client.client_name));
    set_status.provider_client.client_status = g_client->client_status;

    if (g_client->client_status == local_client_status_online)
    {
        std::unique_lock daemon_lock(g_client->daemo_mutex);
        ret = send_msg_to_daemon(g_client->daemon_io, local_registry_msg_service_set_status, &set_status, st_service_set_status_fields, st_service_set_status_size);
        if (IPC_HV_SOA_RET_SUCCESS != ret)
        {
            LOG_PRINT_ERROR("send_msg_to_daemon fail, ret[%d]!", ret);
        }
    }
    else
    {
        LOG_PRINT_ERROR("g_client is offline");
        ret = IPC_HV_SOA_RET_FAIL;
    }

    return ret;
}

int32_t ipc_hv_soa_listener_service_subscribe(ipc_hv_soa_listener_service_t *listener_services, uint32_t listener_services_size)
{
    if (nullptr == g_client || nullptr == listener_services || 0 == listener_services_size)
    {
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    auto it = find_remote_client(g_client->client_id);
    if (nullptr == it)
    {
        LOG_PRINT_ERROR("client[%d] not find", g_client->client_id);
        return IPC_HV_SOA_RET_FAIL;
    }

    st_listen_service subscribe_service = st_listen_service_init_zero;
    subscribe_service.listen_services_count = (pb_size_t)listener_services_size;
    for (size_t i = 0; i < listener_services_size; ++i)
    {
        subscribe_service.listen_services[i].service_id = listener_services[i].service_id;
        subscribe_service.listen_services[i].service_type = listener_services[i].service_type;
        subscribe_service.listen_services[i].service_status = local_service_status_default;

        // save to g_services_map
        std::shared_ptr<ipc_hv_soa_service> service_item = nullptr;
        if (listener_services[i].service_type == E_IPC_HV_SOA_SERVICE_TYPE_EVENT)
        {
            service_item = save_service(listener_services[i].service_id, listener_services[i].service_type, subscribe_service.listen_services[i].service_status, (void *)listener_services[i].u.event_listener_cb, nullptr);
        }
        else
        {
            service_item = save_service(listener_services[i].service_id, listener_services[i].service_type, subscribe_service.listen_services[i].service_status, (void *)listener_services[i].u.method_listener_cb, nullptr);
        }
    }
    subscribe_service.has_listener_client = true;
    subscribe_service.listener_client.client_id = g_client->client_id;
    subscribe_service.listener_client.client_pid = g_client->client_pid;
    std::strncpy(subscribe_service.listener_client.client_name, g_client->client_name.c_str(), sizeof(subscribe_service.listener_client.client_name));
    subscribe_service.listener_client.client_status = g_client->client_status;
    subscribe_service.reg = true;

    if (it->client_status == local_client_status_online)
    {
        std::unique_lock daemon_lock(g_client->daemo_mutex);
        ret = send_msg_to_daemon(g_client->daemon_io, local_registry_msg_listen_service, &subscribe_service, st_listen_service_fields, st_listen_service_size);
        if (IPC_HV_SOA_RET_SUCCESS != ret)
        {
            LOG_PRINT_ERROR("send_msg_to_daemon fail, ret[%d]!", ret);
        }
    }
    else
    {
        LOG_PRINT_ERROR("g_client is offline");
        ret = IPC_HV_SOA_RET_FAIL;
    }

    return ret;
}

int32_t ipc_hv_soa_listener_service_unsubscribe(uint32_t *listener_services, uint32_t listener_services_size)
{
    if (nullptr == g_client || nullptr == listener_services || 0 == listener_services_size)
    {
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    st_listen_service unsubscribe = st_listen_service_init_zero;
    unsubscribe.listen_services_count = (pb_size_t)listener_services_size;
    for (size_t i = 0; i < listener_services_size; ++i)
    {
        auto it = find_service(listener_services[i]);
        if (nullptr == it)
        {
            LOG_PRINT_ERROR("service[%d] not find", listener_services[i]);
            return IPC_HV_SOA_RET_FAIL;
        }
        unsubscribe.listen_services[i].service_id = it->service_id;
        unsubscribe.listen_services[i].service_type = it->service_type;
        // service_status not used
    }
    unsubscribe.has_listener_client = true;
    unsubscribe.listener_client.client_id = g_client->client_id;
    unsubscribe.listener_client.client_pid = g_client->client_pid;
    std::strncpy(unsubscribe.listener_client.client_name, g_client->client_name.c_str(), sizeof(unsubscribe.listener_client.client_name));
    unsubscribe.listener_client.client_status = g_client->client_status;
    unsubscribe.reg = false;

    if (g_client->client_status == local_client_status_online)
    {
        std::unique_lock daemon_lock(g_client->daemo_mutex);
        ret = send_msg_to_daemon(g_client->daemon_io, local_registry_msg_listen_service, &unsubscribe, st_listen_service_fields, st_listen_service_size);
        if (IPC_HV_SOA_RET_SUCCESS != ret)
        {
            LOG_PRINT_ERROR("send_msg_to_daemon fail, ret[%d]!", ret);
        }
    }
    else
    {
        LOG_PRINT_ERROR("g_client is offline");
        ret = IPC_HV_SOA_RET_FAIL;
    }

    return ret;
}

int32_t ipc_hv_soa_method_notify(uint32_t service_id, void *method_req_data, uint32_t method_req_data_len)
{
    if (!g_init_flag)
    {
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    auto it = get_service(service_id);
    if (nullptr == it)
    {
        LOG_PRINT_ERROR("service[%d] not find", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    if (nullptr == it->provider)
    {
        LOG_PRINT_ERROR("service[%d] is no provider", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    if (it->service_type != E_IPC_HV_SOA_SERVICE_TYPE_METHOD || it->service_status != local_service_status_available)
    {
        LOG_PRINT_ERROR("invalid service[%d]", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    uint32_t msg_seqid = it->provider->send_msg_seqid++;
    if (nullptr == method_req_data || method_req_data_len == 0)
    {
        ret = send_msg_to_client(it->provider, g_client->client_id, msg_seqid, E_IPC_HV_SOA_MSG_TYPE_METHOD_NOTIFY, service_id, 0, nullptr);
    }
    else
    {
        ret = send_msg_to_client(it->provider, g_client->client_id, msg_seqid, E_IPC_HV_SOA_MSG_TYPE_METHOD_NOTIFY, service_id, method_req_data_len, method_req_data);
    }

    if (IPC_HV_SOA_RET_SUCCESS != ret)
    {
        LOG_PRINT_ERROR("ipc_hv_soa_method_notify[%d] fail", service_id);
    }

    return ret;
}

int32_t ipc_hv_soa_method_sync(uint32_t service_id, void *method_req_data, uint32_t method_req_data_len, void *method_resp_data, uint32_t *method_resp_data_len, uint32_t timeout_ms)
{
    if (!g_init_flag)
    {
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    auto it = get_service(service_id);
    if (nullptr == it)
    {
        LOG_PRINT_ERROR("service[%d] not find", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    if (nullptr == it->provider)
    {
        LOG_PRINT_ERROR("service[%d] is no provider", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    if (it->service_type != E_IPC_HV_SOA_SERVICE_TYPE_METHOD || it->service_status != local_service_status_available)
    {
        LOG_PRINT_ERROR("invalid service[%d]", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    uint32_t msg_seqid = it->provider->send_msg_seqid++;
    if (nullptr == method_req_data || method_req_data_len == 0)
    {
        ret = send_msg_to_client_sync(it->provider, g_client->client_id, msg_seqid, service_id, 0, nullptr, method_resp_data, method_resp_data_len, timeout_ms);
    }
    else
    {
        ret = send_msg_to_client_sync(it->provider, g_client->client_id, msg_seqid, service_id, method_req_data_len, method_req_data, method_resp_data, method_resp_data_len, timeout_ms);
    }

    return ret;
}

int32_t ipc_hv_soa_method_async(uint32_t service_id, void *method_req_data, uint32_t method_req_data_len, PF_IPC_HV_SOA_METHOD_ASYNC_CB async_cb)
{
    if (!g_init_flag)
    {
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    auto it = get_service(service_id);
    if (nullptr == it)
    {
        LOG_PRINT_ERROR("service[%d] not find", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    if (nullptr == it->provider)
    {
        LOG_PRINT_ERROR("service[%d] is no provider", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    if (it->service_type != E_IPC_HV_SOA_SERVICE_TYPE_METHOD || it->service_status != local_service_status_available)
    {
        LOG_PRINT_ERROR("invalid service[%d]", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    it->service_handler = (void *)async_cb;
    uint32_t msg_seqid = it->provider->send_msg_seqid++;

    ipc_hv_soa_remote_client_data expected_resp_data = {};
    expected_resp_data.service_id = service_id;
    expected_resp_data.msg_type = E_IPC_HV_SOA_MSG_TYPE_METHOD_RESPONSE_ASYNC;
    expected_resp_data.msg_seqid = msg_seqid;
    expected_resp_data.client_id = 0;
    expected_resp_data.is_complete = false;
    expected_resp_data.data_len = 0;
    expected_resp_data.data = nullptr;
    it->provider->send_msg_list.emplace_back(expected_resp_data);

    if (nullptr == method_req_data || method_req_data_len == 0)
    {
        ret = send_msg_to_client(it->provider, g_client->client_id, msg_seqid, E_IPC_HV_SOA_MSG_TYPE_METHOD_REQUEST_ASYNC, service_id, 0, nullptr);
    }
    else
    {
        ret = send_msg_to_client(it->provider, g_client->client_id, msg_seqid, E_IPC_HV_SOA_MSG_TYPE_METHOD_REQUEST_ASYNC, service_id, method_req_data_len, method_req_data);
    }

    if (IPC_HV_SOA_RET_SUCCESS != ret)
    {
        LOG_PRINT_ERROR("ipc_hv_soa_method_async[%d] fail", service_id);
    }
    it->provider->send_msg_list.remove_if([&expected_resp_data](const ipc_hv_soa_remote_client_data &item)
                                          {
                                              return item.service_id == expected_resp_data.service_id &&
                                                  item.msg_seqid == expected_resp_data.msg_seqid;
                                          });

    return ret;
}

static int32_t ipc_hv_soa_inn_sync_complete(uint32_t service_id, uint32_t msg_seqid, void **method_resp_data, uint32_t method_resp_data_len)
{
    if (!g_init_flag)
    {
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    auto it = find_service(service_id);
    if (nullptr == it)
    {
        LOG_PRINT_ERROR("service[%d] not find", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    std::unique_lock<std::mutex> send_msg_lock(it->provider->send_msg_mutex);
    bool getsync = false;

    ipc_hv_soa_remote_client_data *p_resp_data = nullptr;

    for (auto &resp_data : it->provider->send_msg_list)
    {
        if (resp_data.service_id == service_id && resp_data.msg_seqid == msg_seqid)
        {
            p_resp_data = &resp_data;
            getsync = true;
            break;
        }
    }

    if (getsync)
    {
        p_resp_data->is_complete = true;
        p_resp_data->data_len = method_resp_data_len;
        if (method_resp_data_len > 0)
        {
            p_resp_data->data = *method_resp_data;
        }
        else
        {
            p_resp_data->data = nullptr;
        }
        it->provider->send_msg_cond_ret = 0;
        it->provider->send_msg_cond.notify_one();
    }
    else
    {
        LOG_PRINT_ERROR("invalid service[%d]", service_id);
        if (nullptr != *method_resp_data)
        {
            free(*method_resp_data);
        }
        ret = IPC_HV_SOA_RET_ERR_ARG;
    }

    return ret;
}

int32_t ipc_hv_soa_method_complete(ipc_hv_soa_msg_handle_t handle, uint32_t service_id, void *method_resp_data, uint32_t method_resp_data_len)
{
    if (!g_init_flag)
    {
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    uint32_t msg_type = E_IPC_HV_SOA_MSG_TYPE_INVALID;
    if (E_IPC_HV_SOA_MSG_TYPE_METHOD_REQUEST_SYNC == handle.msg_type)
    {
        msg_type = E_IPC_HV_SOA_MSG_TYPE_METHOD_RESPONSE_SYNC;
    }
    else if (E_IPC_HV_SOA_MSG_TYPE_METHOD_REQUEST_ASYNC == handle.msg_type)
    {
        msg_type = E_IPC_HV_SOA_MSG_TYPE_METHOD_RESPONSE_ASYNC;
    }
    else
    {
        LOG_PRINT_ERROR("invalid msg[%d]", handle.msg_type);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    auto it = get_remote_client(handle.client_id);
    if (nullptr == it)
    {
        LOG_PRINT_ERROR("invalid client[%d]", handle.client_id);
        return IPC_HV_SOA_RET_FAIL;
    }

    auto it1 = get_service(service_id);
    if (nullptr == it1)
    {
        LOG_PRINT_ERROR("service[%d] not find", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    if (nullptr == method_resp_data || method_resp_data_len == 0)
    {
        ret = send_msg_to_client(it, g_client->client_id, handle.msg_seqid, msg_type, service_id, 0, nullptr);
    }
    else
    {
        ret = send_msg_to_client(it, g_client->client_id, handle.msg_seqid, msg_type, service_id, method_resp_data_len, method_resp_data);
    }

    if (IPC_HV_SOA_RET_SUCCESS != ret)
    {
        LOG_PRINT_ERROR("ipc_hv_soa_method_complete[%d] fail", service_id);
    }

    return ret;
}

int32_t ipc_hv_soa_inn_trigger_to_client(std::shared_ptr<ipc_hv_soa_service> service, void *event_data, uint32_t event_data_len, std::shared_ptr<ipc_hv_soa_remote_client> client)
{
    if (nullptr == g_client || nullptr == service || nullptr == client)
    {
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    if (nullptr == service->provider)
    {
        LOG_PRINT_ERROR("service[%d] is no provider", service->service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    uint32_t msg_seqid = service->provider->send_msg_seqid++;
    if (nullptr == event_data || 0 == event_data_len)
    {
        ret = send_msg_to_client(client, g_client->client_id, msg_seqid, E_IPC_HV_SOA_MSG_TYPE_EVENT, service->service_id, 0, nullptr);
    }
    else
    {
        ret = send_msg_to_client(client, g_client->client_id, msg_seqid, E_IPC_HV_SOA_MSG_TYPE_EVENT, service->service_id, event_data_len, event_data);
    }

    if (IPC_HV_SOA_RET_SUCCESS != ret)
    {
        LOG_PRINT_ERROR("ipc_hv_soa_inn_trigger_to_client[%d] fail", service->service_id);
    }

    return ret;
}

int32_t ipc_hv_soa_event_trigger(uint32_t service_id, void *event_data, uint32_t event_data_len)
{
    if (!g_init_flag)
    {
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    int32_t ret = IPC_HV_SOA_RET_SUCCESS;

    auto it = get_service_listener(service_id);
    if (nullptr == it)
    {
        LOG_PRINT_ERROR("service[%d] not find", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    if (nullptr == it->provider)
    {
        LOG_PRINT_ERROR("service[%d] is no provider", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    if (it->service_type != E_IPC_HV_SOA_SERVICE_TYPE_EVENT || it->service_status != local_service_status_available)
    {
        LOG_PRINT_ERROR("invalid service[%d]", service_id);
        return IPC_HV_SOA_RET_ERR_ARG;
    }

    for (auto &listener : it->listeners)
    {
        uint32_t msg_seqid = listener.second->send_msg_seqid++;
        if (nullptr == event_data || 0 == event_data_len)
        {
            ret = send_msg_to_client(listener.second, g_client->client_id, msg_seqid, E_IPC_HV_SOA_MSG_TYPE_EVENT, service_id, 0, nullptr);
        }
        else
        {
            ret = send_msg_to_client(listener.second, g_client->client_id, msg_seqid, E_IPC_HV_SOA_MSG_TYPE_EVENT, service_id, event_data_len, event_data);
        }

        if (IPC_HV_SOA_RET_SUCCESS != ret)
        {
            LOG_PRINT_ERROR("ipc_hv_soa_event_trigger[%d] fail", service_id);
        }
    }

    return ret;
}
